from pyspark import SparkConf, SparkContext
import json

json_update = {}

for year in range(2011, 2024):
    json_update[year] = {}
    path = 'hdfs://jiale:9000/zhihu_data/zhihu/' + str(year) + '/' + str(year) + '.txt'
    # 创建 Spark 配置
    conf = SparkConf().setMaster("spark://jiale:7077").setAppName("job01")
    sc = SparkContext(conf=conf)
    lines = sc.textFile(path)

    stopwords = ['的', '了', '你', '在', '我', '有', '也', '都', '就', '不', '可以', '一个', '要', '好', '去', '会', '人', '如果',
                '这个', '与', '等', '很', '就是', '上', '对', '类', '能', '到', '多', '什么', '没有', '说', '很多', '我们', '但',
                '选', '还是', '但是', '更', '想', '大', '因为', '可能', '所以', '比较', '不是', '看', '做', '被', '还', '中', '一定',
                '时候', '了解', '需要', '而', '后', '高', '一些', '年', '个', '这些', '最', '为', '给', '那么', '现在', '分', '不要',
                '问题', '或者', '大家', '知道', '来', '情况', '从', '他', '比如', '非常', '再', '对于', '其实', '时', '所', '相关', '让',
                '比', '又', '还有', '觉得', '吧', '呢', '把', '最好', '最后', '找', '没', '一下', '这样', '下', '真的', '一', '他们', '是',
                '和', '这', '学', '那', '或', '并', '及', '可', '以', '根据', '如何', '已经', '以及', '报', '将', '不同', '主要', '进行', '里',
                '其他', '而且', '过', '才', '一般', '当然', '的话', '之后', '着', '然后', '不会', '不能', '以后', '应该', '小', '读', '这种',
                '们', '啊', '吗', '一样', '来说', '特别', '前', '一点', '它', '开始', '虽然', '当', '得', '怎么', '新', '进入', '这里', '基本', '直接',
                '该', '地', '几个', '跟', '甚至', '能够', '考', '不错', '转', '每个', '只能', '低', '她', '回答', '是否', '只有', '用', '考虑', '只是',
                '发现', '出来', '这么', '当时', '你们', '不过', '啥', '东西', '别人', '结果']

    # 进行词频统计
    word_counts = lines.flatMap(lambda line: line.split(" ")) \
                    .filter(lambda word: word not in stopwords) \
                    .map(lambda word: (word, 1)) \
                    .reduceByKey(lambda a, b: a + b) \
                    .sortBy(lambda x: x[1], ascending=False)

    # 打印词频结果
    for word, count in word_counts.collect():
        if count > 1000:
            json_update[year][word] = count
            print(f"{word}: {count}")

    # 停止 SparkContext
    sc.stop()

with open('../FastAPIProject/json/zhihu_hot.json', 'w', encoding='utf-8') as f:
    json.dump(json_update, f, ensure_ascii=False, indent=4)